package com.foreveross.fetl.util;

import java.lang.reflect.Field;
import java.util.Map;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.AirLineCountVO;

public class AirLineCountUtils {
	private static Logger log = Logger.getLogger("AirLineCountUtils");
	
	public static String apeedRowKey(String channel,String dateStr,String fromCity,String toCity){
		if(null==channel || null==dateStr)
			throw new NullPointerException("channel 或者 dateStr 为 null ");
		dateStr=dateStr.replace("-", "");
		return channel+dateStr+fromCity+toCity;
	}
	
	public static void saveToHbase(AirLineCountVO vo,String tableName){
		if(null==vo || null==vo.getChannel())
			return;
		String rowkey=apeedRowKey(vo.getChannel(),vo.getGrabDate(),vo.getFromCity(),vo.getToCity());
		Put put = new Put(rowkey.getBytes());
		Class<?> clz=vo.getClass();
		Field[] fields=clz.getDeclaredFields();
		for(Field field:fields){
			if("id".equals(field.getName()))
				continue;
			Object value=null;
			field.setAccessible(true);
			try {
				value=field.get(vo);
			} catch (IllegalArgumentException e) {
				log.info(StringUtils.getStackTrace(e));
			} catch (IllegalAccessException e) {
				log.info(StringUtils.getStackTrace(e));
			}
			if(null!=value)
				put.add(Bytes.toBytes("count"), Bytes.toBytes(field.getName()), Bytes.toBytes(value.toString()));
		}
		try {
			HBaseHelper.addRecord(tableName, put);
		} catch (Exception e) {
			log.info(StringUtils.getStackTrace(e));
		}
	}
	
	public static AirLineCountVO getFormHbase(Map<String, Map<String, String>> fcv){
		if(null==fcv || fcv.isEmpty())
			return null;
		if(!fcv.containsKey("count")) 
			return null;
		Map<String, String> map = fcv.get("count");
		AirLineCountVO vo =new AirLineCountVO();
		Class<?> clz=vo.getClass();
		Field[] fields=clz.getDeclaredFields();
		for(Field field:fields){
			if(!map.containsKey(field.getName()))
				continue;
			if("id".equals(field.getName()))
				continue;
			field.setAccessible(true);
			Class<?> type=field.getType();
			try {
				if(type.equals(Integer.class)){
					field.set(vo, Integer.parseInt(map.get(field.getName())));
				}else if(type.equals(Double.class)){
					field.set(vo, Double.parseDouble(map.get(field.getName())));
				}else{
					field.set(vo, map.get(field.getName()));
				}
			} catch (IllegalArgumentException e) {
				log.info(StringUtils.getStackTrace(e));
			} catch (IllegalAccessException e) {
				log.info(StringUtils.getStackTrace(e));
			}
		}
		return vo;
	}
}
